package com.atguigu.hadoop.examples.mapreduce;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.join.CompositeInputFormat;
import org.apache.hadoop.mapreduce.lib.join.TupleWritable;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class Join extends Configured
        implements Tool
{
    public static final String REDUCES_PER_HOST = "mapreduce.join.reduces_per_host";

    static int printUsage()
    {
        System.out.println("join [-r <reduces>] [-inFormat <input format class>] [-outFormat <output format class>] [-outKey <output key class>] [-outValue <output value class>] [-joinOp <inner|outer|override>] [input]* <input> <output>");

        ToolRunner.printGenericCommandUsage(System.out);
        return 2;
    }

    public int run(String[] args)
            throws Exception
    {
        Configuration conf = getConf();
        JobClient client = new JobClient(conf);
        ClusterStatus cluster = client.getClusterStatus();
        int num_reduces = (int)(cluster.getMaxReduceTasks() * 0.9D);
        String join_reduces = conf.get("mapreduce.join.reduces_per_host");
        if (join_reduces != null)
        {
            num_reduces = cluster.getTaskTrackers() *
                    Integer.parseInt(join_reduces);
        }

        Job job = Job.getInstance(conf);
        job.setJobName("join");
        job.setJarByClass(Sort.class);

        job.setMapperClass(Mapper.class);
        job.setReducerClass(Reducer.class);

        Class inputFormatClass = SequenceFileInputFormat.class;

        Class outputFormatClass = SequenceFileOutputFormat.class;

        Class outputKeyClass = BytesWritable.class;
        Class outputValueClass = TupleWritable.class;
        String op = "inner";
        List otherArgs = new ArrayList();
        for (int i = 0; i < args.length; i++) {
            try {
                if ("-r".equals(args[i]))
                    num_reduces = Integer.parseInt(args[(++i)]);
                else if ("-inFormat".equals(args[i]))
                {
                    inputFormatClass = Class.forName(args[(++i)])
                            .asSubclass(InputFormat.class);
                } else if ("-outFormat".equals(args[i]))
                {
                    outputFormatClass = Class.forName(args[(++i)])
                            .asSubclass(OutputFormat.class);
                } else if ("-outKey".equals(args[i]))
                {
                    outputKeyClass = Class.forName(args[(++i)])
                            .asSubclass(WritableComparable.class);
                } else if ("-outValue".equals(args[i]))
                {
                    outputValueClass = Class.forName(args[(++i)])
                            .asSubclass(Writable.class);
                } else if ("-joinOp".equals(args[i]))
                    op = args[(++i)];
                else
                    otherArgs.add(args[i]);
            }
            catch (NumberFormatException except) {
                System.out.println("ERROR: Integer expected instead of " + args[i]);
                return printUsage();
            } catch (ArrayIndexOutOfBoundsException except) {
                System.out.println("ERROR: Required parameter missing from " + args[(i - 1)]);

                return printUsage();
            }

        }

        job.setNumReduceTasks(num_reduces);

        if (otherArgs.size() < 2) {
            System.out.println("ERROR: Wrong number of parameters: ");
            return printUsage();
        }

        FileOutputFormat.setOutputPath(job, new Path(
                (String)otherArgs
                        .remove(otherArgs
                                .size() - 1)));
        List plist = new ArrayList(otherArgs.size());
        for (Object s : otherArgs) {
            plist.add(new Path(s.toString()));
        }

        job.setInputFormatClass(CompositeInputFormat.class);
        job.getConfiguration().set("mapreduce.join.expr",
                CompositeInputFormat.compose(op, inputFormatClass,
                        (Path[])plist
                                .toArray(new Path[0])));

        job.setOutputFormatClass(outputFormatClass);

        job.setOutputKeyClass(outputKeyClass);
        job.setOutputValueClass(outputValueClass);

        Date startTime = new Date();
        System.out.println("Job started: " + startTime);
        int ret = job.waitForCompletion(true) ? 0 : 1;
        Date end_time = new Date();
        System.out.println("Job ended: " + end_time);
        System.out.println("The job took " +
                (end_time
                        .getTime() - startTime.getTime()) / 1000L + " seconds.");
        return ret;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new Join(), args);
        System.exit(res);
    }
}